package com.nari.interplatform.mq;

import com.alibaba.fastjson.JSONObject;
import com.nari.common.core.constant.Constants;
import com.nari.common.core.utils.SpringUtils;
import com.nari.common.log.enums.RstType;
import com.nari.common.log.service.AsyncLogService;
import com.nari.interplatform.domain.RstRequest;
import com.nari.interplatform.domain.SwReq;
import com.nari.interplatform.domain.SwResponse;
import com.nari.interplatform.service.IRstRequestService;
import com.nari.interplatform.utils.HttpUtil;
import com.nari.interplatform.utils.MsgUtil;
import com.nari.interplatform.utils.XmlUtil;
import com.nari.system.api.domain.SysRstLog;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;


@Component
public class Consumer {
    /**
     * 消息生产者
     */
    private DefaultMQProducer producer = SpringUtils.getBean(Producer.class).getProducer();
    private MqConfig mqConfig = SpringUtils.getBean(MqConfig.class);
    @Autowired
    private IRstRequestService rstRequestService ;//= SpringUtils.getBean(RstRequestServiceImpl.class)
    @Autowired
    private AsyncLogService asyncLogService;
    /**
     * 消费者实体对象
     */
    private DefaultMQPushConsumer consumer;
    /**
     * 消费者组
     */
   // public  final String CONSUMER_GROUP = mqConfig.getConsumerGroup();
    private Logger LOGGER = LoggerFactory.getLogger(Consumer.class);
    /**
     * 通过构造函数 实例化对象
     */
    public Consumer() throws MQClientException {
       // SpringUtils.getBean(MqConfig.class);
        consumer = new DefaultMQPushConsumer(mqConfig.getConsumerGroup());
       // System.out.println("namesrvAddr======="+SpringUtils.getBean(MqConfig.class).getNamesrvAddr());
        //System.out.println("namesrvAddr======="+mqConfig.getNamesrvAddr());
        consumer.setNamesrvAddr(mqConfig.getNamesrvAddr());
        //消费模式:一个新的订阅组第一次启动从队列的最后位置开始消费 后续再启动接着上次消费的进度开始消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        //订阅主题和 标签（ * 代表所有标签)下信息
        //System.out.println("topic======="+mqConfig.getTopic());
        //设置集群消费模式
        consumer.setMessageModel(MessageModel.CLUSTERING);
        consumer.subscribe(mqConfig.getTopicReq(), "*");
        consumer.setMaxReconsumeTimes(3);
        //设置消费超时时间(分钟)
        consumer.setConsumeTimeout(1000L * 60);
        // //注册消费的监听 并在此监听中消费信息，并返回消费的状态信息
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            // msgs中只收集同一个topic，同一个tag，并且key相同的message
            // 会把不同的消息分别放置到不同的队列中
                    if (CollectionUtils.isEmpty(msgs)) {
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }

                    for (MessageExt msg : msgs) {

                    if (!processor(msg)){
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }else{
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }

                }

            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });

        consumer.start();
        LOGGER.info("消费者 启动成功=======");
    }
    /**
     * 消息处理，第3次处理失败后，修改状态，查明原因，待补推
     * @param message
     * @return
     */
    private boolean processor(MessageExt message)  {
        SwReq swReq = new SwReq();
        SwResponse swResponse = new SwResponse();
        RstRequest rstRequest= new RstRequest();
        SysRstLog sysRstLog= new SysRstLog();
        String body = "";
       /* try {*/
            try {
                body = new String(message.getBody(), Constants.GBK);
                LOGGER.info("Consumer-获取消息主题topic为={" + message.getTopic() + "}, 消费消息为={" + body + "}");
                //读取队列消息日志入库
                sysRstLog = fitSysRstLog(body);
                sysRstLog.setType(RstType.LOG_003_REQGET.getLogType());
                sysRstLog.setStatus(Constants.RTS_LOG_SUCCESS);
                sysRstLog.setcData(body);
                asyncLogService.saveRstLog(sysRstLog);

            }catch (Exception e){
                e.printStackTrace();

                if(message.getReconsumeTimes()>=3){
                    sysRstLog = fitSysRstLog(body);
                    sysRstLog.setType(RstType.LOG_003_REQGET.getLogType());
                    sysRstLog.setStatus(Constants.RTS_LOG_FAIL);
                    sysRstLog.setcData(body);
                    asyncLogService.saveRstLog(sysRstLog);
                    return true;
                }
                return false;
            }
        try {
            //解析请求
            String encryptUsed =Constants.ENCRYPT_USED;
            String compressUsed = Constants.COMPRESS_USED;
            swReq = MsgUtil.dealMsg(body,encryptUsed,compressUsed);
            //签名完成 日志入库
            sysRstLog = new SysRstLog();
            sysRstLog.setTranId(swReq.getSwReqHeader().getTranId());
            sysRstLog.setTranSEQ(swReq.getSwReqHeader().getTranSeq());
            sysRstLog.setType(RstType.LOG_004_REQSIGN.getLogType());
            sysRstLog.setStatus(Constants.RTS_LOG_SUCCESS);
            sysRstLog.setcData(swReq.getBody());
            asyncLogService.saveRstLog(sysRstLog);
           } catch (Exception e){
                e.printStackTrace();

                if(message.getReconsumeTimes()>=3){

                    sysRstLog.setType(RstType.LOG_004_REQSIGN.getLogType());
                    sysRstLog.setStatus(Constants.RTS_LOG_FAIL);
                    sysRstLog.setcData(swReq.getBody());
                    asyncLogService.saveRstLog(sysRstLog);
                    return true;
                }
                return false;
            }

            //组装生成报文xml
            String xml = XmlUtil.toXML(swReq);
            //拼接完成日志入库

            rstRequest.setRequest(xml);
            String tranSeq = swReq.getSwReqHeader().getTranSeq();
            rstRequest.setTranSeq(tranSeq);
            if(!isExtSwReq(tranSeq)){
                rstRequest.setStatus(Constants.RTS_REQUEST_DEALING);
                rstRequestService.insertRstRequest(rstRequest);//读取消息入库
            }
            sysRstLog.setType(RstType.LOG_005_REQPOST.getLogType());
            sysRstLog.setStatus(Constants.RTS_LOG_SUCCESS);
            sysRstLog.setcData(xml);
            asyncLogService.saveRstLog(sysRstLog);
            //发送报文请求税务
            /*//正式环境
            //String endpoint = "https://sbf.shandong.chinatax.gov.cn:8443/extsService/services/extsService";
            ExtsServicePortTypeProxy proxy = new ExtsServicePortTypeProxy(Constants.WSDL_URL);
            System.setProperty("javax.net.ssl.trustStore", "F:\\sdzhsw\\RuoYi-Vue\\jssecacerts");
            String result = proxy.doService(xml);*/
            //测试环境
            Map<String,Object> paramMap = new HashMap<String,Object>();
            paramMap.put("url", Constants.WSDL_URL);
            paramMap.put("inputData", xml);
           //  System.out.println("消费者 启动成功=======");
           LOGGER.info("请求税务xml========"+xml);
            //调用测试环境返回报文
            String result = HttpUtil.doPost("http://121.89.206.47:8080/IntfTest/intfTest", paramMap);
            LOGGER.info("税务返回xml========"+result);
            //调用税务完成 日志入库
            sysRstLog.setType(RstType.LOG_006_GETREP.getLogType());
            sysRstLog.setStatus(Constants.RTS_LOG_SUCCESS);
             sysRstLog.setcData(result);
            asyncLogService.saveRstLog(sysRstLog);
            try {
                //解析报文
                swResponse = XmlUtil.fromXML(result, SwResponse.class);
                //读取接口配置 设置压缩、签名启用状态
                String encryptUsed =Constants.ENCRYPT_USED;
                String compressUsed = Constants.COMPRESS_USED;
                swResponse.getSwResponseHeader().getSwReqContentControl().getCompress().setUsed(compressUsed);
                swResponse.getSwResponseHeader().getSwReqContentControl().getEncrypt().setUsed(encryptUsed);
                swResponse = MsgUtil.dealRespMsg(swResponse);
                //完成报文解析入库及记录日志
                rstRequest.setResponse(swResponse.getBody());
                rstRequest.setRestCode(swResponse.getSwResponseHeader().getRtnCode());
                rstRequestService.updateRstRequest(rstRequest);
                sysRstLog.setType(RstType.LOG_007_VERIFY.getLogType());
                sysRstLog.setStatus(Constants.RTS_LOG_SUCCESS);
                sysRstLog.setcData(swResponse.getBody());
                sysRstLog.setRemark(swResponse.getSwResponseHeader().getSwRtnMsg().getResson());
                asyncLogService.saveRstLog(sysRstLog);

            }catch (Exception e){
                    e.printStackTrace();

                    if(message.getReconsumeTimes()>=3){
                        //"消息重试已达最大次数，修改状态，查明原因，待补推。{}",message.getMsgId());
                        rstRequest.setStatus(Constants.RTS_REQUEST_FAIL);
                        rstRequestService.updateRstRequest(rstRequest);
                        sysRstLog.setType(RstType.LOG_007_VERIFY.getLogType());
                        sysRstLog.setStatus(Constants.RTS_LOG_FAIL);
                        sysRstLog.setcData(swResponse.getBody());
                        asyncLogService.saveRstLog(sysRstLog);
                        return true;
                    }
                    return false;
                }
            try{
            //写队列 发送
            String msg = MsgUtil.respJsonString(swResponse);
            Message respMessage = new Message(mqConfig.getTopicResp(),"*", msg.getBytes("GBK"));
            LOGGER.info("发送响应队列消息========"+respMessage.toString());
            producer.sendOneway(respMessage);
            //发送返回消息 记录日志
                sysRstLog.setType(RstType.LOG_008_REPPUT.getLogType());
                sysRstLog.setStatus(Constants.RTS_LOG_SUCCESS);
                sysRstLog.setcData(msg);
                asyncLogService.saveRstLog(sysRstLog);
                rstRequest.setStatus(Constants.RTS_REQUEST_SUCCESS);
                rstRequest.setResponse(msg);
                rstRequestService.updateRstRequest(rstRequest);
            }catch (Exception e){
                e.printStackTrace();

                if(message.getReconsumeTimes()>=3){
                    //"消息重试已达最大次数，修改状态，查明原因，待补推。{}",message.getMsgId());
                    rstRequest.setStatus(Constants.RTS_REQUEST_FAIL);
                    rstRequestService.updateRstRequest(rstRequest);
                    sysRstLog.setType(RstType.LOG_008_REPPUT.getLogType());
                    sysRstLog.setStatus(Constants.RTS_LOG_FAIL);
                    asyncLogService.saveRstLog(sysRstLog);
                    return true;
                }
                return false;
            }
      /*  }catch (Exception e){
            e.printStackTrace();
            if(message.getReconsumeTimes()>=3){
                //"消息重试已达最大次数，修改状态，查明原因，待补推。{}",message.getMsgId());

                return true;
            }
            return false;
        }*/
        //处理消息发送请求
        return true;
    }
    private boolean isExtSwReq(String tranSeq){
        RstRequest rstRequest = rstRequestService.checkRstRequestKeyUnique(tranSeq);
        if(rstRequest!=null&&rstRequest.getId()>0){
            return true;
        }else{
            return false;
        }
    }

    /**
     * 生成请求信息
     * @param msg
     * @return
     */
    private RstRequest fitRstRequest(String msg){
        JSONObject msgJson = new JSONObject();
        msgJson = JSONObject.parseObject(msg).getJSONObject("Request");
        String tranId = msgJson.get("tran_id") == null ? null : msgJson.getString("tran_id");// 调用服务的交易编码
        String tranName = msgJson.get("tran_name") == null ? null : msgJson.getString("tran_name");// 接口名称
        String tranSeq = msgJson.get("tran_seq") == null ? "" : msgJson.getString("tran_seq");// 交易流水号
        String taxpayer = msgJson.get("taxpayer") == null ? "" : msgJson.getString("taxpayer");// 纳税主体
        String source = msgJson.get("source") == null ? "" : msgJson.getString("source");// 请求来源系统
        String tranDateReq = msgJson.get("tran_date_req") == null ? "" : msgJson.getString("tran_date_req");// 交易发生日期，格式yyyyMMdd
        String tranTimeReq = msgJson.get("tran_time_req") == null ? "" : msgJson.getString("tran_time_req");// 交易发生时间，格式HH24MISSFFF
        String body = msgJson.get("body") == null ? "" : msgJson.getString("body");// 业务报文
        RstRequest rstRequest = new RstRequest();
        rstRequest.setTranId(tranId);
        rstRequest.setTranName(tranName);
        rstRequest.setTranSeq(tranSeq);
        rstRequest.setTaxpayer(taxpayer);
        rstRequest.setSource(source);
        rstRequest.setTranDateReq(tranDateReq);
        rstRequest.setTranTimeReq(tranTimeReq);
        rstRequest.setRequest(body);
        rstRequest.setStatus(Constants.RTS_REQUEST_DEALING);
        return rstRequest;
    }

    /**
     * 根据请求信息生成日志信息
     * @param msg
     * @return
     */
    private SysRstLog fitSysRstLog(String msg){

        SysRstLog sysRstLog = new SysRstLog();
        JSONObject msgJson = new JSONObject();
        msgJson = JSONObject.parseObject(msg).getJSONObject("Request");
        String tranId = msgJson.get("tran_id") == null ? null : msgJson.getString("tran_id");// 调用服务的交易编码
        String tranName = msgJson.get("tran_name") == null ? null : msgJson.getString("tran_name");// 接口名称
        String tranSeq = msgJson.get("tran_seq") == null ? "" : msgJson.getString("tran_seq");// 交易流水号
        String taxpayer = msgJson.get("taxpayer") == null ? "" : msgJson.getString("taxpayer");// 纳税主体

        sysRstLog.setTranId(tranId);
        sysRstLog.setTranName(tranName);
        sysRstLog.setTranSEQ(tranSeq);
        sysRstLog.setTaxpayer(taxpayer);

        return sysRstLog;
    }

}
